package com.ruoyi.flink.wc.test;

import java.util.List;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.flink.util.read.ClickHouseRead;
import com.ruoyi.flink.util.write.ClickHouseWrite;
import com.ruoyi.flink.wc.entity.Student;
import com.ruoyi.flink.wc.util.RocketMQSource;

/**
 * @author Acer
 * 这个测试, 从rocketmq获取年龄数据, 将这个年龄段的小朋友查出来,找到,叫王二丫的小朋友, 给王二丫,加上一岁.
 */
public class ClickhostRead {

	public static void main(String[] args) throws Exception {
		

        // System.setProperty("rocketmq.client.logLevel", "ERROR");
        // 参数中获取值
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        // 获取 rocketmq的地址端口  "192.168.1.125:9876,192.168.1.125:9875"
        String namesrvaddr = parameterTool.get("namesrvaddr", "192.168.1.123:9876");
        // 获取主题(topic)
        String topic = parameterTool.get("topic", "flink-test");
        // 获取分组(group)
        String group = parameterTool.get("flinkGroup", "flink-group");
        // 创建流式执行环境
        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度(默认就是以)
        env.setParallelism(1);
        // 创建表环境
        //  StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

       
        String sql = "select name,age,sex from default.student";
       
       
        // 获取mq中的参数值   // 这是拿到了单个的值     
//        DataStreamSource<String> mqSource = env.addSource(new RocketMQSource(namesrvaddr, topic, group));
//        SingleOutputStreamOperator<Integer> wordAndOne =  mqSource.map(new MapFunction<String, Integer>() {
//            @Override
//            public Integer map(String value) {
//                JSONObject json = JSONObject.parseObject(value);
//                return json.getInteger("age");
//      
//            }
//        });

        //
  
    DataStreamSource<String> mqSource = env.addSource(new RocketMQSource(namesrvaddr, topic, group));
    SingleOutputStreamOperator<Integer> wordAndOne =  mqSource.map(new MapFunction<String, Integer>() {
		private static final long serialVersionUID = 1L;

		@Override
        public Integer map(String value) {
            JSONObject json = JSONObject.parseObject(value);
            return json.getInteger("age");
  
        }
    });
      
     sql = "select name,age,sex from  default.student where age=?";
     
     
    // mqSource.addSink(new ClickHoustReadUtil(sql));
     

	}
}
